package com.huaxin.dss2.message;

import com.huaxin.dss2.message.bean.DataMsgBean;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.BlockingDeque;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import javax.annotation.PostConstruct;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.BoundHashOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

/**
 * Created by HX-119 on 2014/7/25.
 */
@Component
@Slf4j
public class MessageCache extends Thread{



    @Autowired
    private RedisTemplate<String, String> template;

    private BoundHashOperations<String, String, Object> stationOps;
    private BoundHashOperations<String, String, Object> dataOps;
    private BoundHashOperations<String, String, Object> dataStateOps;
    public static BoundHashOperations<String, String, Object> stationStaticOps;

    private BlockingDeque<DataMsgBean> queue = new LinkedBlockingDeque<>();

    private Map<String,DataMsgBean> dataMsgBeanLocalCache = new ConcurrentHashMap<>();//修改为相对线程安全的ConcurrentHashMap
    private Map<String,Long> orgCodeLastDataTimeMap = new ConcurrentHashMap<>();//修改为相对线程安全的ConcurrentHashMap
    private int dsize = 0;

    @PostConstruct
    private void init() {

        // 序列化 - 后续可能会用到
//        template.setHashValueSerializer(new JacksonJsonRedisSerializer<>(Object.class));
//        template.setHashKeySerializer(new JacksonJsonRedisSerializer<>(Object.class));
//        template.setKeySerializer(new JacksonJsonRedisSerializer<>(Object.class));

        stationOps = template.boundHashOps("station");

        dataOps = template.boundHashOps("data");
        stationStaticOps = template.boundHashOps("station");
        dataStateOps = template.boundHashOps("dataState");
        //2019/6/6取消缓存过期时间
        //dataOps.expire(10, TimeUnit.SECONDS);
        //dataStateOps.expire(10, TimeUnit.SECONDS);
        this.start();
    }

    public void putToCache(DataMsgBean dataMsgBean) {
        putToLocalCache(dataMsgBean);
    }

    private void putToLocalCache(DataMsgBean dataMsgBean) {
        try {
            queue.put(dataMsgBean);
            log.debug("put message to local queue");
        } catch (InterruptedException e) {
            log.error(e.getMessage(),e);
        }
    }

    /**
     * 存入redis
     * @param dataMsgBeans 数据集合
     */
    public void putToRemoteCache(Collection<DataMsgBean> dataMsgBeans) {
        dsize = dataMsgBeans.size();
        for(DataMsgBean dataMsgBean : dataMsgBeans){
            try {
                Map<String, Object> data = new HashMap<>();
                data.put("stationId", dataMsgBean.getStationId());
                if (!StringUtils.isEmpty(dataMsgBean.getErrorInfo()) && dataMsgBean.getErrorInfo().equals("DATAOK")) {
                    data.put("dataTime", dataMsgBean.getDataTime());
                }else{
                    Map<String,Object> map = (Map<String, Object>) stationOps.entries().get(dataMsgBean.getStationId());
                    if(null!=map){
                        Object dataTime = map.get("dataTime");
                        if(null != dataTime){
                            data.put("dataTime",dataTime);
                        }
                    }
                }
                data.put("errorInfo", dataMsgBean.getErrorInfo());
                stationOps.put(dataMsgBean.getStationId(), data);
                dataOps.putAll(dataMsgBean.getDataMap());
                dataStateOps.putAll(dataMsgBean.getStateMap());
                //dateOps.put(dataMsgBean.getStationId());
            } catch (Exception e) {
                log.error("写入缓存时发生错误," + e.getMessage(), e);
            }
        }
    }

    private boolean isNew(String orgCode,Long dataTime) {
        if(dataTime>System.currentTimeMillis()+600000){
            log.warn("[orgCode:{}]数据时间[{}]大于当前时间十分钟以上,禁止放入缓存.",orgCode,dataTime);
            return false;
        }

        Long lastDataTime = orgCodeLastDataTimeMap.get(orgCode);

        if(null==lastDataTime){
            return true;
        }else{
            return dataTime-lastDataTime>=0;
        }
    }

    @Override
    public void run() {
        while (true){
            try {
                DataMsgBean dataMsgBean =queue.take();
                if(isNew(dataMsgBean.getStationId(),dataMsgBean.getDataTime())){
                    dataMsgBeanLocalCache.put(dataMsgBean.getStationId(),dataMsgBean);
                    orgCodeLastDataTimeMap.put(dataMsgBean.getStationId(),dataMsgBean.getDataTime());
                }
            } catch (Exception e) {
                log.error(e.getMessage(),e);
            }
        }
    }

    /**
     * redis 定时任务
     */
    private final AtomicBoolean isFlushing = new AtomicBoolean(false);
    public void flushToRemoteCache(){
        //处理线程存入redis时map线程不安全和错误赋值的问题
        if(!isFlushing.compareAndSet(false,true)){
            log.warn("上次任务未执行完毕,跳过");
            return;
        }

        try {
            if (CollectionUtils.isEmpty(dataMsgBeanLocalCache)){
                log.warn("队列中暂无缓存数据");
                return;
            }
            //清空缓存队列存入redis中
            //Map<String,DataMsgBean> currentCache = new ConcurrentHashMap<>(dataMsgBeanLocalCache);
            //currentCache.forEach((key,value)-> dataMsgBeanLocalCache.remove(key,value));
            Map<String,DataMsgBean> currentCache;
            synchronized(dataMsgBeanLocalCache){
                currentCache = new HashMap<>(dataMsgBeanLocalCache);
            }

            //2019.5.15 对dataMsgBeanLocalCache进行锁处理 by lindy
            for ( Map.Entry<String,DataMsgBean> entry: currentCache.entrySet()) {
                dataMsgBeanLocalCache.remove(entry.getKey(),entry.getValue());
            }
            if(currentCache.values().size()>0){
                putToRemoteCache(currentCache.values());
                log.info("put redis length : {}", currentCache.values().size());
            }
        } finally {
            isFlushing.set(false);
        }

    }
}
